package com.atguigu.edu.realtime.app.dwd.db;


import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 课程评价事实表
 */
public class DwdInteractionCommentInfo {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(5);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka的主题中读取数据  创建动态表---kafka连接器
        String groupId = "dwd_interaction_comment_group";
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL(groupId));
        //tableEnv.executeSql("select * from topic_db").print();

        //TODO 4.过滤出评论数据
        Table commentTable = tableEnv.sqlQuery("select\n" +
                "`data`['id']id,\n" +
                "`data`['user_id']user_id,\n" +
                "`data`['course_id'] course_id,\n" +
                "`data`['review_txt'] review_txt,\n" +
                "`data`['review_stars'] review_stars,\n" +
                " ts\n" +
                " from topic_db\n" +
                " where `table`='review_info'\n" +
                " and `type`='insert'");
        tableEnv.createTemporaryView("review_info",commentTable);
        tableEnv.executeSql("select * from review_info").print();

        //TODO 5.将评论数据的结果写到kafka主题---upsert-kafka
        //5.1 创建动态表和要写入的主题进行映射
        tableEnv.executeSql("\n" +
                "CREATE TABLE dwd_interaction_comment(\n" +
                "id String,\n" +
                "user_id String,\n" +
                "course_id String,\n" +
                "review_txt String,\n" +
                "review_stars String,\n" +
                "ts String,\n" +
                "PRIMARY KEY (id) NOT ENFORCED\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_interaction_comment"));
        //5.2 写入
        tableEnv.executeSql("insert into dwd_interaction_comment select * from review_info");

    }
}
